package com.i360day.redis.listener;

import com.i360day.redis.annotation.RedisExpireListener;
import com.i360day.redis.annotation.RedisKeyExpireListener;
import com.i360day.redis.emuns.MatchingType;
import com.i360day.redis.listener.keyexpire.RedisKeyExpireActuator;
import com.i360day.redis.listener.keyexpire.RedisKeyExpireListenerAdapter;
import com.i360day.redis.lock.RedisLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;

/**
 * <p> @Author liju.z <p>
 *
 * <p> @Description redis key值过期通知 <p>
 *
 * <p> @Date  16:06 <p>
 *
 * <p> @Param  <p>
 *
 * <p> @return  <p>
 **/
final class RedisKeyExpireMessageListener extends KeyExpirationEventMessageListener {

    private final Logger logger = LoggerFactory.getLogger(RedisKeyExpireMessageListener.class);

    private RedisKeyExpireActuator redisKeyExpireActuator;

    private static final Topic KEYEVENT_EXPIRED_TOPIC = new PatternTopic("__keyevent@*__:expired");

    public RedisKeyExpireMessageListener(RedisMessageListenerContainer redisMessageListenerContainer, RedisKeyExpireActuator redisKeyExpireActuator) {
        super(redisMessageListenerContainer);
        this.redisKeyExpireActuator = redisKeyExpireActuator;
    }

    @Override
    protected void doRegister(RedisMessageListenerContainer listenerContainer) {
        listenerContainer.addMessageListener(this, KEYEVENT_EXPIRED_TOPIC);
    }

    /**
     * <p> @Author liju.z <p>
     *
     * <p> @Description 获取key <p>
     *
     * <p> @Date  8:52 <p>
     *
     * <p> @Param [clazzKey] <p>
     *
     * <p> @return [clazzKey] <p>
     **/
    private String[] getKeys(Class<?> clazzKey) {
        RedisExpireListener redisExpireListener = AnnotationUtils.findAnnotation(clazzKey, RedisExpireListener.class);
        if (redisExpireListener != null) {
            List<String> classNameList = new ArrayList<>();
            for (Class<?> clazz : redisExpireListener.value()) {
                classNameList.add(clazz.getName());
            }
            return classNameList.stream().toArray(String[]::new);
        } else {
            RedisKeyExpireListener redisKeyExpireListener = AnnotationUtils.findAnnotation(clazzKey, RedisKeyExpireListener.class);
            Assert.isTrue(redisKeyExpireListener != null, String.format("%s必须有RedisExpireListener RedisKeyExpireListener", clazzKey.getName()));
            return redisKeyExpireListener.value();
        }
    }

    private boolean isLock(Class<?> clazzKey) {
        RedisExpireListener redisExpireListener = AnnotationUtils.findAnnotation(clazzKey, RedisExpireListener.class);
        if (redisExpireListener != null) {
            return redisExpireListener.lock();
        } else {
            RedisKeyExpireListener redisKeyExpireListener = AnnotationUtils.findAnnotation(clazzKey, RedisKeyExpireListener.class);
            return redisKeyExpireListener.lock();
        }
    }

    /**
     * <p> @Author liju.z <p>
     *
     * <p> @Description 获取匹配条件 <p>
     *
     * <p> @Date  8:52 <p>
     *
     * <p> @Param [clazzKey] <p>
     *
     * <p> @return [clazzKey] <p>
     **/
    private MatchingType getCondition(Class<?> clazzKey) {
        RedisExpireListener redisExpireListener = AnnotationUtils.findAnnotation(clazzKey, RedisExpireListener.class);
        if (redisExpireListener != null) {
            return redisExpireListener.condition();
        } else {
            RedisKeyExpireListener redisKeyExpireListener = AnnotationUtils.findAnnotation(clazzKey, RedisKeyExpireListener.class);
            Assert.isTrue(redisKeyExpireListener != null, String.format("%s必须有RedisExpireListener RedisKeyExpireListener", clazzKey.getName()));
            return redisKeyExpireListener.condition();
        }
    }

    @Override
    protected void doHandleMessage(Message message) {
        //回调父类，通知出spring事件。自带命名空间
        super.doHandleMessage(message);
        //获取到过期的key值
        final String _expireKey = new String(message.getBody());
        Collection<RedisKeyExpireListenerAdapter> list = redisKeyExpireActuator.getRedisKeyExpireListenerAdapterList();
        for (RedisKeyExpireListenerAdapter redisKeyExpireListenerAdapter : list) {
            //获取到自定义key
            String[] keys = getKeys(redisKeyExpireListenerAdapter.getClass());
            //上锁操作
            boolean lock = isLock(redisKeyExpireListenerAdapter.getClass());
            //获取到匹配条件类型
            MatchingType matchingType = getCondition(redisKeyExpireListenerAdapter.getClass());
            //查找
            long keyCount = Arrays.asList(keys).stream().filter(f -> matchingType.compare(_expireKey, f)).count();
            if (keyCount > 0) {
                if (lock) {
                    try {
                        RedisLock.instance(_expireKey, 1).call(() -> {
                            Thread.sleep(1000L);//默认等待1秒，使别的线程获取锁失败
                            redisKeyExpireActuator.invoke(redisKeyExpireListenerAdapter, _expireKey);
                        });
                    } catch (Throwable throwable) {
                        logger.warn("过期key值【{}】获取锁失败...{}", _expireKey, throwable.getMessage());
                    }
                } else {
                    redisKeyExpireActuator.invoke(redisKeyExpireListenerAdapter, _expireKey);
                }
            }
        }
    }
}
